package org.lql.chaos.kafka.example.callback;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.NonNull;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 16:41
 * @description:
 **/
public class CustomProducerCallback<K, V> implements ListenableFutureCallback<SendResult<K, V>> {

	private static final Logger log = LoggerFactory.getLogger(CustomProducerCallback.class);

	private final K key;
	private final V value;

	public CustomProducerCallback(K key, V value) {
		this.key = key;
		this.value = value;
	}

	@Override
	public void onSuccess(SendResult<K, V> result) {
		if (result != null && result.getRecordMetadata() != null) {
			RecordMetadata metadata = result.getRecordMetadata();
			log.info("消息发送成功！topic={}, partition={}, offset={}, key={}, value={}",
					metadata.topic(), metadata.partition(), metadata.offset(),
					key, value);
		} else {
			log.warn("消息发送成功，但 RecordMetadata 为空，key={}, value={}", key, value);
		}
	}

	@Override
	public void onFailure(@NonNull Throwable ex) {
		log.error("消息发送失败！key={}, value={}, error={}", key, value, ex.getMessage(), ex);
		// 这里可以做重试或报警等处理
	}
}
